
package com.infindo.frame.data;

import java.util.Vector;

@SuppressWarnings("unchecked")
public class AsyncProcessor {
	private final Vector queue = new Vector(); //remove is O(n)
	private final Object lock = this;
	private boolean disposed = false;

	/**
	 * Constructs an AsyncProcessor and starts the specified number
	 * of worker threads.
	 */
	public AsyncProcessor(int numberOfWorkerThreads) {
		for (int i = 0; i < numberOfWorkerThreads; i++) {
			new Worker().start();
		}
	}

	/**
	 * Disposes of this AsyncProcessor. No new jobs can be added, and worker 
	 * threads will terminate once the current job queue is drained.
	 */
	public void dispose() {
		synchronized (lock) {
			disposed = true;
			lock.notifyAll();
		}
	}


	/**
	 * Adds a job who's run() method will be invoked asynchronously by a worker thread.
	 */
	public void enqueueJob(Runnable job) {
		if (job == null) throw new IllegalArgumentException("job is null");
		synchronized (lock) {
			if (disposed) {
				throw new IllegalStateException("Attempt to add job to disposed AsyncProcessor queue.");
			}
			queue.addElement(job);
			lock.notify();
		}
	}
	
	public synchronized boolean asynCancelHandler(int dataHandlerNo)
	{
		for(int i=0;i<queue.size();i++)
		{
			DataHandler mHandler;
			mHandler = (DataHandler)queue.elementAt(i);
			if(mHandler.getDataHandlerNo()==dataHandlerNo)
			{
				queue.removeElementAt(i);
				i--;
				return true;
			}
		}
		return false;
	}
	

	private class Worker extends Thread {
		
		Worker() {
			super();
		}

		public void run() {
			try{
				for (;;) {
					// Get a job from the queue
					Runnable job = null;
					synchronized (lock) 
					{
						while (job == null) {
							if (queue.size() > 0) {
								job = (Runnable)queue.firstElement();
								queue.removeElementAt(0);
							}
							else if (disposed) {
								return;
							}
							else {
								try { lock.wait(); }
								catch (InterruptedException e) { }
							}
						}
					}
					// Run it
					boolean done = false;
					try {
							job.run();
							done = true;
					} catch (Throwable e) {
						e.printStackTrace();
					}
					finally {
						if (!done) {
							// Some kind of exception/error. Allow it to be caught by the default
							// exception handler, but start another worker to replace this one.
							new Worker().start();
						}
					}
				}
			} catch (Throwable e) {
				e.printStackTrace();
			}
		}
	}
}
